package com.fjwt.gz.service.activity.platform.acte017;

import com.alibaba.fastjson2.JSONObject;
import com.fjwt.gz.core.cache.RedisUtil;
import com.fjwt.gz.core.constants.ApiCodeEnum;
import com.fjwt.gz.core.constants.Constants;
import com.fjwt.gz.core.exception.BizException;
import com.fjwt.gz.core.utils.ApiStatisticsLogger;
import com.fjwt.gz.core.utils.SpringBeansUtil;
import com.fjwt.gz.db.dto.FansActivityDTO;
import com.fjwt.gz.service.activity.BaseActivityService;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.commons.lang3.ObjectUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import static com.fjwt.gz.core.cache.RedisUtil.getStringRedisTemplate;

@Component
class ConsumeThread implements ApplicationRunner,Runnable {
    static final Logger log = LoggerFactory.getLogger(ConsumeThread.class);

    private AtomicReference<ExecutorService> workerRef;
    private final AtomicBoolean isRunning = new AtomicBoolean(false);
    private volatile long lastTaskTime = 0;
    private final long outTime = 300_000;
    private final long redisOutTime = 300_000;
    private final int queueThreshold = 10;
    private static final String keyPrefix = Constants.REDIS_ACTIVITY.FILENAME + "ACTE017";
    private static final String queue = keyPrefix + ":queue";
    private static final String pendingQueue = keyPrefix + ":pendingQueue";
    private static ApiStatisticsLogger logger = ApiStatisticsLogger.getInstance();

    @Value("${redis.isConsumeThread}")
    private boolean isConsumeThread;

    // 添加线程工厂
    protected static final ThreadFactory namedThreadFactory =
            new ThreadFactoryBuilder().setNameFormat(keyPrefix + "consumer-%d").build();

    // 修改线程池创建方式
    protected ExecutorService createExecutor() {
        return Executors.newSingleThreadExecutor(namedThreadFactory);
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        if (isConsumeThread) {
            workerRef = new AtomicReference<>(createExecutor());
            submitTask();
        }else {
            workerRef = new AtomicReference<>(null);
        }
    }

    public void check(FansActivityDTO fansActivityDTO){
        Long fansId = fansActivityDTO.getFansId();

        RedisTemplate stringRedisTemplate = RedisUtil.getStringRedisTemplate();

        //判断用户是否存在消息队列中
        List<String> userQueueList = stringRedisTemplate.opsForList().range(queue, 0, -1);

        // 记录消息队列当前排队人数
        ApiStatisticsLogger.apiStatisticsLogger.info("当前队列中的人数：{}", ObjectUtils.isEmpty(userQueueList) ? 0 : userQueueList.size());

        if (userQueueList !=null  && !userQueueList.isEmpty()) {
            for (String userQueueStr : userQueueList) {
                FansActivityDTO fansActivityQueue = JSONObject.parseObject(userQueueStr, FansActivityDTO.class);
                if (fansId.equals(fansActivityQueue.getFansId()) && fansActivityDTO.getActivityId().equals(fansActivityQueue.getActivityId())) {
                    log.info("用户fansId：{}已在排队队列中", fansId);
                    throw new BizException(ApiCodeEnum.LITE_USER_COUPON, "您正在抢券中，请耐心等待！");
                }
            }
        }

        //判断用户是否在pending队列中
        List<String> userPendingQueueList = stringRedisTemplate.opsForList().range(pendingQueue, 0, -1);
        if (userPendingQueueList !=null  && !userPendingQueueList.isEmpty()) {
            for (String userPendingStr :userPendingQueueList){
                FansActivityDTO fansActivityPending = JSONObject.parseObject(userPendingStr, FansActivityDTO.class);
                if (fansId.equals(fansActivityPending.getFansId()) && fansActivityDTO.getActivityId().equals(fansActivityPending.getActivityId())){
                    log.info("用户fansId：{}已在pending队列中", fansId);
                    throw new BizException(ApiCodeEnum.LITE_USER_COUPON, "卡券正在发放中，请耐心等待！");
                }
            }
        }
        //说明用户不在排队消息队列中 需要查询是在pending队列中
        if (userQueueList != null && userQueueList.size() >= queueThreshold){
            log.info("用户fansId：{}参与活动时队列已满", fansId);
            throw new BizException("排队人数太多了，请稍后再试！");
        }else {
            // 记录用户加入队列时间
            long joinQueueTime = System.currentTimeMillis();
            fansActivityDTO.setJoinQueueTime(joinQueueTime);
            long userJoinQueueTime = joinQueueTime - fansActivityDTO.getStartTime();
            logger.logRequest("userJoinQueueTime", userJoinQueueTime);

            // 把用户放在排队队列中
            stringRedisTemplate.opsForList().leftPush(queue, JSONObject.toJSONString(fansActivityDTO));
            log.info("用户放在排队队列中，信息是：{}", JSONObject.toJSONString(fansActivityDTO));
            if (isConsumeThread) {
                submitTask();
            }
        }
    }

    public void run() {
        BaseActivityService activityService = SpringBeansUtil.getBean("ACTE017Service", BaseActivityService.class);
        // 类成员变量
        RedisTemplate<String, String> stringRedisTemplate = getStringRedisTemplate();
        log.info("开始执行任务");
        while (!Thread.currentThread().isInterrupted()) {
            try {
                // 1：消费者从队列未弹出消息并推送到 pending 队列整个过程是原子性的 最多阻塞 5 秒超过 5 秒后还没有消息则返回 null
                String item = stringRedisTemplate.opsForList().rightPopAndLeftPush(queue, pendingQueue, 5, TimeUnit.SECONDS);
                if (item == null) {
                    log.info("等待消息 ...");
                    continue;
                }
                Long fansId = 0L;
                Long activityId = 0L;
                try {
                    // 记录用户队列等待时间
                    FansActivityDTO fansActivityDTO = JSONObject.parseObject(item, FansActivityDTO.class);
                    long startConsumeTime = System.currentTimeMillis();
                    logger.logRequest("userWaitingTime", startConsumeTime - fansActivityDTO.getJoinQueueTime());

                    activityId = fansActivityDTO.getActivityId();
                    fansId = fansActivityDTO.getFansId();
                    log.info("fansId：{}，开始进行消费", fansId);
                    activityService.handleActivityAwardBO(fansActivityDTO);

                    long endTime = System.currentTimeMillis();
                    // 记录用户队列消费时间
                    logger.logRequest("userConsumeTime", endTime - startConsumeTime);

                    // 记录用户参与活动总时长
                    long consumeTime = endTime - fansActivityDTO.getStartTime();
                    log.info("fansId：{}，消费完毕！本次耗时：{}ms",fansId, consumeTime);

                    logger.logRequest("/api/activity/takePartIn", consumeTime);
                } catch (Exception e) {
                    log.error("消费异常：{}", e.getMessage());
                    RedisUtil.setString(Constants.REDIS_ACTIVITY.FILENAME + activityId + ":consumeError:" + fansId, e.getMessage(), redisOutTime, TimeUnit.MILLISECONDS);

                    e.printStackTrace();
                }
                // 3：消费成功从 pending 队列删除记录相当于确认消费
//                stringRedisTemplate.opsForList().remove(pendingQueue, 0, item);//此为匹配删除
                stringRedisTemplate.opsForList().trim(pendingQueue, 1, 0);//此为全删除
            } catch (Exception e) {
                log.error("队列监听异常：{}", e.getMessage());
                break;
            }finally {
                isRunning.set(false);
            }
        }
        log.info("退出消费");
    }

    /**
     * 提交新任务
     */
    public void submitTask() {
        if (isConsumeThread) {
            // 原子性地获取或创建线程池
            workerRef.updateAndGet(w -> (w == null || w.isShutdown()) ? createExecutor() : w);
            lastTaskTime = System.currentTimeMillis(); // 更新最后任务时间
            log.info("有任务提交，启动线程");
            startWorkerIfNeeded();
        }
    }

    /**
     * 启动工作线程（如果未运行）
     */
    private void startWorkerIfNeeded() {
        if (isConsumeThread) {
            if (isRunning.compareAndSet(false, true)) {
                workerRef.get().execute(this);
                log.info("启动线程，执行任务");
            }
        }
    }

    /**
     * 停止工作线程
     */
    private void shutdownWorker() {
        workerRef.getAndUpdate(w -> {
            if (w != null && !w.isShutdown()) {
                log.info("关闭线程");
                w.shutdown();
                try {
                    if (!w.awaitTermination(5, TimeUnit.SECONDS)) {
                        w.shutdownNow();
                    }
                    log.info("关闭线程成功");
                } catch (InterruptedException e) {
                    w.shutdownNow();
                    Thread.currentThread().interrupt();
                } finally {
                    isRunning.set(false);
                }
            }
            return null; // 清空引用
        });
    }
}